﻿#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <signal.h> //信号处理
#include <pthread.h> //线程
#include <semaphore.h>
#include "threadpool.h"
#include "queue.h"


int flag = 0;


//生产者线程
void* producer(void* arg)
{
	struct threadpool* p = (struct threadpool*)arg;//强转
	while (flag == 0)
	{
		struct exam k;
		memset(&k, 0, sizeof(k));					//可以用针持续指向数据，也可以用一个新的数据变量给如队列q->que中！！！
		for (int i = 0; i < p->queue_length_max; i++)								
		{
			sem_wait(&p->semw);//阻塞
			if (flag == 1)
			{
				break;
			}
			pthread_mutex_lock(&p->mutex);//互斥锁
			printf("给队列里输入两个值:\n");
			scanf("%d", &k.data1);
			scanf("%d", &k.data2);

			push(p->que, &k);											
			printf("入队数据是：%d %d\n", k.data1,k.data2);

			pthread_mutex_unlock(&p->mutex);//释放锁
		}

		for (int i = 0; i < p->queue_length_max; i++)												
		{
			sem_post(&p->semr);//唤醒消费者线程										
		}


	}
	printf("生产者线程退出\n");
}

//消费者线程
void* consumer(void* arg)
{
	int count = 0;
	struct threadpool* p = (struct threadpool*)arg;

	while (flag == 0)
	{

		sem_wait(&p->semr);//阻塞
		if (flag == 1)
		{
			break;
		}
		pthread_mutex_lock(&p->mutex);//互斥锁


		printf("\n消费者线程---%ld号在处理数据：\n", pthread_self() % 100);//获取线程id
		count = size(p->que);//发送队列头结点求大小
		printf("队列长度是：%d\n", count);
		front(p->que);//查找队列头结点的第一个数据
		pop(p->que);//出队


		pthread_mutex_unlock(&p->mutex);//释放锁

		sleep(1);

	}

	printf("消费者线程---%ld号退出\n", pthread_self() % 100);
}

void fun(int sig)
{
	flag = 1;
}

int main(int argc, const char* argv[])
{
	signal(SIGINT, fun);

	int ret = 0;
	int threads = 0;
	int queue_length = 0;
	threads = 3;//控制消费者线程数量
	queue_length = 5;//队列长度控制的是一个生产者生产的数量						

	printf("线程池中消费者线程的数量:%d\n", threads);
	printf("队列长度:%d\n", queue_length);

	// 创建线程池---多消费者线程
	struct threadpool* p = create_threadpool(threads, queue_length, consumer);
	p->que = create_queue();//队列的头结点
	// 创建生产者线程
	pthread_t tid1 = 0;
	p->queue_length_max = queue_length;
	sem_init(&p->semw, 0, queue_length);//初始化信号量			
	sem_init(&p->semr, 0, 0);//初始化信号量		
	pthread_mutex_init(&p->mutex, NULL);//初始化互斥锁
	p->que = create_queue();
	ret = pthread_create(&tid1, NULL, producer, p);//激活生产者线程
	if (ret != 0) {
		perror("pthread_create");
	}

	while (flag == 0)
	{
		sleep(1);
	}
	sem_post(&p->semw);//唤醒生产者线程
	//应该唤醒所有消费者线程退出!!!!
	for (int i = 0; i < threads; i++)
	{
		sem_post(&p->semr);//唤醒消费者线程
	}


	pthread_join(tid1, NULL);//等待生产者线程结束
	destroy_threadpool(p, threads);//销毁线程池
	destroy_queue(p->que);//释放队列
	printf("已释放队列及销毁线程池\n");

	return 0;

}
